package com.camemax.rabbitmq.work_mode.fair;

import com.camemax.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

// *********************
// ** @Date: 2021-06-09
// ** @Author: Camemax
// ** @Description: Consumer Template for RabbitMQ Consumer in work mode [ by fair ].
// *********************
public class WorkFairConsumerRunnable implements Runnable
{
    private ConnectionFactory connFactory = RabbitMQUtils.createFactory();
    private Connection conn = null;
    private Channel channel = null;
    private String queueName;
    private Integer sleepTime; // watch Thread sleeps in Fair mode easily.

    public WorkFairConsumerRunnable(String queueName){
        this(queueName, 1000);
    }

    public WorkFairConsumerRunnable(String queueName, Integer sleepTime){
        this.queueName = queueName;
        this.sleepTime = sleepTime;
    }

    @Override
    public void run() {
        String consumerName = Thread.currentThread().getName(); // Use thread's name as RabbitMQ Consumers' name.
        try {
            conn = connFactory.newConnection(consumerName);
            channel = conn.createChannel();
            channel.basicQos(1); // Confirm Consumer consume messages' count each time.
            channel.basicConsume(
                    queueName,
                    Boolean.FALSE, // Close auto-response ark.
                    (consumer, delivery) -> {
                        System.out.println(consumerName + " accepts message: " + new String(delivery.getBody(), StandardCharsets.UTF_8));
                        // Confirm the ack's operation.
                        // @params: deliveryTag => Ack index.
                        // @params: multiple => All messages which deliveryTag is less than Ack index , will be responsed in batch.
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                        try {
                            TimeUnit.MILLISECONDS.sleep(sleepTime);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    },
                    (consumer) -> System.out.println(consumerName + " accepts failed!")
            );
            System.out.println(consumerName + " loaded successfully!Consuming Start!");
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            RabbitMQUtils.close(channel , conn);
        }
    }
}
